﻿using System;
using System.Collections.Generic;
using System.Linq;
using Elasticsearch.Synchronize.Common.Enum;
using Elasticsearch.Synchronize.Domain.Aggregates.SynchronizeAggregate;
using Elasticsearch.Synchronize.Domain.Interface;
using HYService.BLModels.Event;
using HYService.RabbitMQ;
using MongoDB.Driver;
using Nest;
using Polly;

namespace Elasticsearch.Synchronize.DomainEventHandler.Processor
{
    public class EventJoinProcessor : BaseProcessor<HYEventJoin>
    {
        public EventJoinProcessor(SynchronizeItem synchronizeItem, ISynchronizeItemRepository repository)
            : base(synchronizeItem, repository)
        {
        }

        public override List<HYEventJoin> Query()
        {
            var filter = new JsonFilterDefinition<HYEventJoin>(_synchronizeItem.JsonFilter);
            var result = _mongoCollection.Find(filter).ToList();
            return result;
        }

        public override void Pour()
        {
            var data = this.Query();

            if (data.Count < _synchronizeItem.Threshold)
            {
                return;
            }

            var pageIndex = 1;
            var pageSize = 10000;
            while (true)
            {
                try
                {
                    var temp = data.Skip((pageIndex - 1) * pageSize).Take(pageSize).ToList();
                    if (!temp.Any())
                    {
                        try
                        {
                            Policy.Handle<Exception>().WaitAndRetry(5, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (exception, timespan, retryCount, context) =>
                            {
                            }).Execute(() =>
                            {
                                var response = _elasticClient.Count<HYEventJoin>(s => s.Query(q => q.Type(t => t.Name(_synchronizeItem.ElasticsearchTypeName))));
                                if (!response.IsValid)
                                {
                                    throw response.OriginalException;
                                }

                                if (response.Count != data.Count)
                                {
                                    throw new Exception("The count of docs in elasticsearch is not match the count of docs in mongodb");
                                }

                                _synchronizeItem.State = SyncState.Tail;
                                _repository.Update(_synchronizeItem.Identity, _synchronizeItem);
                            });
                        }
                        catch
                        {
                            _synchronizeItem.Pour();
                        }

                        break;
                    }

                    try
                    {
                        Policy.Handle<Exception>().WaitAndRetry(5, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (exception, timespan, retryCount, context) =>
                        {
                        }).Execute(() =>
                        {
                            var response = _elasticClient.IndexMany(temp, _synchronizeItem.ElasticsearchIndexName, _synchronizeItem.ElasticsearchTypeName);

                            if (!response.IsValid)
                            {
                                throw response.OriginalException;
                            }

                            if (pageIndex == 1)
                            {
                                _synchronizeItem.State = SyncState.Pouring;
                                _repository.Update(_synchronizeItem.Identity, _synchronizeItem);
                            }

                            pageIndex++;
                        });
                    }
                    catch
                    {
                        continue;
                    }
                }
                catch
                {
                    continue;
                }
            }
        }
    }
}